package PassengerVehicle

import java.io.{BufferedReader, FileReader}

import org.apache.spark.sql.SparkSession


/**
  * 省内省外的载客量统计
  * 2月份
  * IN,550244
  * OUT,9968
  * 3月份
  * IN,561236
  * OUT,5383
  * 4月份
  * IN,677367
  * OUT,6491
  *
  */
object ProvinInOutByChu {
  val path = "D:/projects/spark/src/main/resource/demo.csv"
  def main(args: Array[String]): Unit = {
    //匹配规则
    val reader = new FileReader("D:/projects/spark/src/main/resource/result.txt")
    val bufferReader = new BufferedReader(reader)
    val dict = bufferReader.readLine()

    bufferReader.close()
    reader.close()

    val spark = SparkSession.builder().master("local[4]").appName("mainProcess").getOrCreate()

    val frame = spark.read.format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat").option("header", true).option("delimiter", ",").load(path)

    // IN 省内  OUT 省外
    val result = frame.rdd.map(item => {
      val id = item.getString(0)
      val xl = item.getString(1)
      val ddz = item.getString(2)
      if (matchDDZ(xl, ddz, dict)) {
        ("IN", 1)
      } else {
        ("OUT", 1)
      }
    }).reduceByKey((x, y) => x + y)
    result.foreach(x => {
      println(x._1 + "," + x._2)
    })
  }

  def matchDDZ(xl: String, ddz: String, dict: String): Boolean = {
    val reg1 = "[0-9]"
    val reg2 = "（[\\u4e00-\\u9fa5]*）"
    val reg3 = "[(][0-9\\u4e00-\\u9fa5]*[)]"
    val reg4 = "服务站|下道车|机场|国际机场|车站|汽车站|长途车站|长途汽车站|中心站|中心|高速学生|高速|补|路口"

    val xllen = xl.split("-").length
    if (dict.indexOf(ddz.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      return true
    } else if (xllen == 2 && dict.indexOf(xl.split("-")(1).replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      return true
    } else if (xllen == 1 && dict.indexOf(xl.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      return true
    } else {
      return false
    }
  }

}
